package com.cellulam.framework.mq.listener;

import com.aliyun.openservices.ons.api.Admin;
import com.aliyun.openservices.ons.api.Message;
import com.aliyun.openservices.ons.api.PropertyKeyConst;
import com.aliyun.openservices.ons.api.PropertyValueConst;
import com.google.common.collect.Maps;
import com.cellulam.framework.core.utils.JacksonUtils;
import com.cellulam.framework.core.base.ResourceId;
import com.cellulam.framework.core.spring.SmartEnvironment;
import com.cellulam.framework.mq.ActionResult;
import com.cellulam.framework.mq.MessageFeature;
import com.cellulam.framework.mq.RocketMqConfig;
import com.cellulam.framework.mq.RocketMqConsumerConfig;
import com.cellulam.framework.mq.exceptions.MqConsumeInterruptException;
import com.cellulam.properties.SystemProperties;
import lombok.Getter;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;

import java.util.List;
import java.util.Map;
import java.util.Properties;
import java.util.function.Function;

@Slf4j
public abstract class BaseAbstractMessageListener<C extends Admin>
        implements MqListener {

    @Autowired
    private SystemProperties systemProperties;

    private RocketMqConfig mqConfig;

    private MqConsumeMessageMode messageMode;

    private String topic;

    private C consumer;

    private Map<String, Processor<?>> processors;

    protected BaseAbstractMessageListener(MqConsumeMessageMode messageMode) {
        this.messageMode = messageMode;
    }

    protected abstract C createConsumer(Properties properties);

    @Override
    public void start() {
        this.consumer.start();
    }

    @Override
    public final void initConsumer(RocketMqConsumerConfig consumerConfig) {
        this.createConsumer(consumerConfig);
        this.processors = Maps.newHashMap();
        this.registerProcessors();
        log.info("注册消息处理器完成, groupId: {}, processors size: {}",
                this.mqConfig.getGroupId(),
                this.processors.size());
    }

    protected abstract void registerProcessors();

    protected final <T extends ResourceId> void addProcessor(MqFeatureProcess<T> featureProcess) {

        Class<T> resourceClass = featureProcess.getResourceClass();
        MqProcessor mqProcessor = featureProcess.getClass().getAnnotation(MqProcessor.class);
        List<String> tags = SmartEnvironment.env.resolveIfNeed(mqProcessor.tags());

        this.addProcessors(tags, featureProcess, resourceClass);
    }

    private <T extends ResourceId> void addProcessors(List<String> tags, MqFeatureProcess<T> process, Class<T> resourceClass) {
        tags.forEach(x -> this.addProcessor(x,
                messageFeature -> process.process(messageFeature),
                resourceClass,
                process.getClass().getName()));
    }

    private String getProcessorKey(String groupId, String topic, String tag) {
        return String.format("%s#%s#%s",
                groupId,
                topic,
                tag);
    }

    private <T extends ResourceId> void addProcessor(String tag,
                                                     Function<MessageFeature<T>, ActionResult> processor,
                                                     Class<T> resourceClass,
                                                     String processorName) {
        String key = this.getProcessorKey(this.mqConfig.getGroupId(), this.topic, tag);
        this.processors.put(key, new Processor<>(processor, resourceClass));
        log.info("注册消息处理器[{}], groupId: {}, topic: {}, tag: {}, processor: {}",
                key,
                this.mqConfig.getGroupId(),
                this.topic,
                tag,
                processorName);
    }

    protected ActionResult consumeMessage(Message message) {
        try {
            log.info("Receive message group: {}, topic: {}, tag: {}, msgId: {}, msgKey: {}",
                    this.mqConfig.getGroupId(),
                    message.getTopic(),
                    message.getTag(),
                    message.getMsgID(),
                    message.getKey());
            String key = this.getProcessorKey(this.mqConfig.getGroupId(), message.getTopic(), message.getTag());
            Processor processor = this.processors.get(key);

            if (processor == null) {
                return this.processByDefaultProcessor(key, message);
            }

            return processor.process(this.mqConfig.getGroupId(), message);

        } catch (MqConsumeInterruptException e) {
            log.error("{}消费{}终止, topic:{}, tag: {}, msgKey: {}, body: {}",
                    this.mqConfig.getGroupId(),
                    message.getMsgID(),
                    message.getTopic(),
                    message.getTag(),
                    message.getKey(),
                    new String(message.getBody()), e);
            return ActionResult.SUCCESS;
        } catch (Exception e) {
            log.error("{}消费{}异常, topic:{}, tag: {}, msgKey: {}, body: {}",
                    this.mqConfig.getGroupId(),
                    message.getMsgID(),
                    message.getTopic(),
                    message.getTag(),
                    message.getKey(),
                    new String(message.getBody()), e);
            return ActionResult.FAIL;
        }
    }

    protected ActionResult processByDefaultProcessor(String processorName, Message message) {
        log.warn("未找到处理器: {}, msgId: {}, msgKey: {}",
                processorName, message.getMsgID(), message.getKey());
        return ActionResult.SUCCESS;
    }

    private synchronized void createConsumer(RocketMqConsumerConfig consumerConfig) {
        if (this.consumer != null) {
            log.warn("消费者已创建");
            return;
        }

        this.mqConfig = this.initMqConfig(consumerConfig);
        this.topic = consumerConfig.getTopic();

        Properties properties = this.mqConfig.toMqProperties();
        properties.put(PropertyKeyConst.MessageModel, this.messageMode.equals(MqConsumeMessageMode.BROADCASTING)
                ? PropertyValueConst.BROADCASTING
                : PropertyValueConst.CLUSTERING);
        properties.put(PropertyKeyConst.ConsumeThreadNums, String.valueOf(consumerConfig.getConcurrentConsumers()));
        properties.put(PropertyKeyConst.SuspendTimeMillis, String.valueOf(consumerConfig.getSuspendTimeMillis()));
        properties.put(PropertyKeyConst.MaxReconsumeTimes, String.valueOf(consumerConfig.getMaxReConsumeTimes()));
        properties.put(PropertyKeyConst.ConsumeTimeout, String.valueOf(consumerConfig.getConsumeTimeout()));

        this.consumer = this.createConsumer(properties);
        log.info("创建消费者, groupId: {}",
                this.mqConfig.getGroupId());
    }

    protected final C getConsumer() {
        return this.consumer;
    }

    private RocketMqConfig initMqConfig(RocketMqConsumerConfig consumerConfig) {
        return RocketMqConfig.builder()
                .accessKey(this.systemProperties.getAliMqAccessKey())
                .secretKey(this.systemProperties.getAliMqSecretKey())
                .nameSrv(this.systemProperties.getAliMqNameSrv())
                .groupId(consumerConfig.getGroupId())
                .build();
    }

    @Getter
    private static class Processor<T extends ResourceId> {
        public Processor(Function<MessageFeature<T>, ActionResult> processor, Class<T> resourceClass) {
            this.processor = processor;
            this.resourceClass = resourceClass;
        }

        public MessageFeature<T> convertToFeature(String groupId, Message message) {
            T body = JacksonUtils.toObjByGenericType(new String(message.getBody()), this.resourceClass);
            return new MessageFeature<>(groupId, message, body);
        }

        private Function<MessageFeature<T>, ActionResult> processor;
        private Class<T> resourceClass;

        public ActionResult process(String groupId, Message message) {
            return this.getProcessor().apply(this.convertToFeature(groupId, message));
        }
    }

}
